#include <unistd.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include "iomanager.h"
#include "macro.h"
#include "log.h"

enum EpollCtlOp
{
};

static std::ostream &operator<<(std::ostream &os, const EpollCtlOp &op)
{
    switch ((int)op)
    {
#define XX(ctl) \
    case ctl:   \
        return os << #ctl;
        XX(EPOLL_CTL_ADD);
        XX(EPOLL_CTL_MOD);
        XX(EPOLL_CTL_DEL);
#undef XX
    default:
        return os << (int)op;
    }
}

static std::ostream &operator<<(std::ostream &os, EPOLL_EVENTS events)
{
    if (!events)
    {
        return os << "0";
    }
    bool first = true;
#define XX(E)          \
    if (events & E)    \
    {                  \
        if (!first)    \
        {              \
            os << "|"; \
        }              \
        os << #E;      \
        first = false; \
    }
    XX(EPOLLIN);
    XX(EPOLLPRI);
    XX(EPOLLOUT);
    XX(EPOLLRDNORM);
    XX(EPOLLRDBAND);
    XX(EPOLLWRNORM);
    XX(EPOLLWRBAND);
    XX(EPOLLMSG);
    XX(EPOLLERR);
    XX(EPOLLHUP);
    XX(EPOLLONESHOT);
    XX(EPOLLET);
#undef XX
    return os;
}

IOManager::FdContext::EventContext &IOManager::FdContext::getEventContext(Event event)
{
    switch (event)
    {
    case IOManager::READ:
        return IOManager::FdContext::read;
    case IOManager::WRITE:
        return IOManager::FdContext::write;
    default:
        SYLAR_ASSERT2(false, "getcontext");
    }
    throw std::invalid_argument("getcontext invalid event");
}

void IOManager::FdContext::resetEventContext(EventContext &ctx)
{
    ctx.callback = nullptr;
    ctx.fiber.reset();
    ctx.scheduler = nullptr;
}

// 触发事件
void IOManager::FdContext::triggerEvent(IOManager::Event event)
{
    // 待触发的事件必须被注册过
    SYLAR_ASSERT(events & event);

    // 清除该事件,表示不再关注该事件了,也就是说,如果想持续关注某个socket fd的读写事件
    // 那么每次触发事件之后都要重新添加
    events = (Event)(events & ~event);

    // 调度对应的协程
    EventContext &ctx = getEventContext(event);
    if (ctx.callback)
    {
        ctx.scheduler->schedule(ctx.callback);
    }
    else
    {
        ctx.scheduler->schedule(ctx.fiber);
    }

    resetEventContext(ctx);
    return;
}

IOManager::IOManager(size_t threads, bool use_caller, const std::string &name)
    : Scheduler(threads, use_caller, name)
{
    m_epfd = epoll_create(5000);
    SYLAR_ASSERT(m_epfd > 0);

    // 创建pipe,获取m_tickleFds[2],其中m_tickleFds[0]是管道的读端,m_tickleFds[1]是管道的写端
    int ret = pipe(m_tickleFds);
    SYLAR_ASSERT(!ret);

    // 注册pipe读句柄的可读事件,⽤于tickle调度协程,通过epoll_event.data.fd保存描述符
    epoll_event event;
    memset(&event, 0, sizeof(event));
    event.events = EPOLLIN | EPOLLET;
    event.data.fd = m_tickleFds[0];

    // 非阻塞方式,配合边缘触发
    ret = fcntl(m_tickleFds[0], F_SETFL, O_NONBLOCK);
    SYLAR_ASSERT(!ret);

    // 将管道的读描述符加⼊epoll多路复⽤,如果管道可读,idle中的epoll_wait会返回
    ret = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickleFds[0], &event);
    SYLAR_ASSERT(!ret);

    // 重置socket句柄上下文的容器大小
    contextResize(32);

    // 启动调度器 这⾥直接开启了Schedluer,也就是说IOManager创建即可调度协程
    Scheduler::start();
}

IOManager::~IOManager()
{
    // 停止调度器,等所有的任务调度都执行完了再返回
    stop();

    close(m_epfd);
    close(m_tickleFds[0]);
    close(m_tickleFds[1]);

    for (size_t i = 0; i < m_fdContexts.size(); i++)
    {
        if (m_fdContexts[i])
            delete m_fdContexts[i];
    }
}

// 重置socket句柄上下文的容器大小
void IOManager::contextResize(size_t size)
{
    m_fdContexts.resize(size);

    for (size_t i = 0; i < m_fdContexts.size(); i++)
    {
        if (!m_fdContexts[i])
        {
            m_fdContexts[i] = new FdContext;
            m_fdContexts[i]->fd = i;
        }
    }
}

// 添加事件
int IOManager::addEvent(int fd, Event event, std::function<void()> cb)
{
    // 找到fd对应的FdContext,如果不存在就分配一个
    FdContext *fd_ctx = nullptr;
    RWMutexType::ReadLock lock(m_mutex);
    if ((int)m_fdContexts.size() > fd)
    {
        fd_ctx = m_fdContexts[fd];
        lock.unlock();
    }
    else
    {
        lock.unlock();
        RWMutexType::WriteLock lock(m_mutex);
        contextResize(fd * 1.5);
        fd_ctx = m_fdContexts[fd];
    }

    // 同一个fd不允许重复添加相同的事件
    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if (SYLAR_UNLIKELY(fd_ctx->events & event))
    {
        LOG(ERROR, "addEvent assert fd = %d event = %d fd_ctx.event = %d", fd, (EPOLL_EVENTS)event, (EPOLL_EVENTS)fd_ctx->events);
        SYLAR_ASSERT(!(fd_ctx->events & event));
    }

    // 将新的事件加入到epoll_wait,使用epoll_event的私有指针存储FdContext的位置
    int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
    epoll_event epevent;
    epevent.events = EPOLLET | fd_ctx->events | event;
    epevent.data.ptr = fd_ctx;

    int ret = epoll_ctl(m_epfd, op, fd, &epevent);
    if (ret)
    {
        LOG(ERROR, "epoll_ctl( %d, %d,%d, %d,): %d ( %d ) (%s) fd_ctx->events = %d",
            m_epfd, (EpollCtlOp)op, fd, (EPOLL_EVENTS)epevent.events, ret, errno, strerror(errno), (EPOLL_EVENTS)fd_ctx->events);
        return -1;
    }

    // 待执行的IO事件加1
    ++m_pendingEventCount;

    // 找到这个fd的event事件对应的EventContext,对其中的scheduler,cb,fiber进行赋值
    fd_ctx->events = (Event)(fd_ctx->events | event);
    FdContext::EventContext &event_ctx = fd_ctx->getEventContext(event);
    SYLAR_ASSERT(!event_ctx.scheduler && !event_ctx.fiber && !event_ctx.callback);

    // 赋值scheduler和回调函数,如果回调函数为空,则把当前协程当成回调执行体
    event_ctx.scheduler = Scheduler::GetThis();
    if (cb)
    {
        event_ctx.callback.swap(cb);
    }
    else
    {
        event_ctx.fiber = Fiber::GetThis();
        SYLAR_ASSERT2(event_ctx.fiber->getState() == Fiber::RUNNING, "state=" << event_ctx.fiber->getState());
    }
    return 0;
}

// 删除事件
bool IOManager::delEvent(int fd, Event event)
{
    // 找到对应的fdcontext
    RWMutexType::ReadLock lock(m_mutex);
    if ((int)m_fdContexts.size() <= fd)
    {
        return false;
    }

    FdContext *fd_ctx = m_fdContexts[fd];
    lock.unlock();

    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if (SYLAR_UNLIKELY(!fd_ctx->events & event))
    {
        return false;
    }

    // 清除指定的事件,表示不关心这个事件了,如果清除之后结果为0,则从epoll_wait中删除该文件描述符
    Event new_events = (Event)(fd_ctx->events & ~event);
    int op = new_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.data.ptr = fd_ctx;

    int ret = epoll_ctl(m_epfd, op, fd, &epevent);
    if (ret)
    {
        LOG(ERROR, "epoll_ctl( %d, %d, %d, %d,):%d, (%d)(%s)", m_epfd, (EpollCtlOp)op, fd, (EPOLL_EVENTS)epevent.events, ret, errno, strerror(errno));
        return false;
    }

    // 待执行事件数减1
    --m_pendingEventCount;

    // 重置该fd对应的event上下文
    fd_ctx->events = new_events;
    FdContext::EventContext &event_ctx = fd_ctx->getEventContext(event);
    fd_ctx->resetEventContext(event_ctx);

    return true;
}

// 取消事件
bool IOManager::cancelEvent(int fd, Event event)
{
    // 找到fd对应的FdContext
    RWMutexType::ReadLock lock(m_mutex);
    if ((int)m_fdContexts.size() <= fd)
    {
        return false;
    }

    FdContext *fd_ctx = m_fdContexts[fd];
    lock.unlock();

    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if (SYLAR_UNLIKELY(!(fd_ctx->events & event)))
    {
        return false;
    }

    // 删除事件
    // 清除指定的事件,表示不关⼼这个事件了,如果清除之后结果为0,则从epoll_wait中删除该⽂件描述符
    Event new_events = (Event)(fd_ctx->events & ~event);
    int op = new_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;

    epoll_event epevent;
    epevent.events = EPOLLET | new_events;
    epevent.data.ptr = fd_ctx;

    int ret = epoll_ctl(m_epfd, op, fd, &epevent);
    if (ret)
    {
        LOG(ERROR, "epoll_ctl( %d, %d, %d, %d,):%d, (%d)(%s)", m_epfd, (EpollCtlOp)op, fd, (EPOLL_EVENTS)epevent.events, ret, errno, strerror(errno));
        return false;
    }

    // 删除之前触发一次的事件
    fd_ctx->triggerEvent(event);
    // 活跃事件数减一
    --m_pendingEventCount;
    return true;
}

// 取消所有事件
bool IOManager::cancelAll(int fd)
{
    // 找到fd对应的FdContext
    RWMutexType::ReadLock lock(m_mutex);
    if ((int)m_fdContexts.size() <= fd)
    {
        return false;
    }

    FdContext *fd_ctx = m_fdContexts[fd];
    lock.unlock();

    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if (SYLAR_UNLIKELY(!(fd_ctx->events)))
    {
        return false;
    }

    // 删除全部事件
    int op = EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = 0;
    epevent.data.ptr = fd_ctx;

    int ret = epoll_ctl(m_epfd, op, fd, &epevent);
    if (ret)
    {
        LOG(ERROR, "epoll_ctl( %d, %d, %d, %d,):%d, (%d)(%s)", m_epfd, (EpollCtlOp)op, fd, (EPOLL_EVENTS)epevent.events, ret, errno, strerror(errno));
        return false;
    }

    // 触发全部已经注册的事件
    if (fd_ctx->events & READ)
    {
        fd_ctx->triggerEvent(READ);
        --m_pendingEventCount;
    }
    if (fd_ctx->events & WRITE)
    {
        fd_ctx->triggerEvent(WRITE);
        --m_pendingEventCount;
    }

    SYLAR_ASSERT(fd_ctx->events == 0);
    return true;
}

// 返回当前的IOManager
IOManager *IOManager::GetThis()
{
    return dynamic_cast<IOManager *>(Scheduler::GetThis());
}

/***
 * 通知调度协程,也就是Scheduler::run()从idle中退出
 * Scheduler::run()每次从idle中退出之后,都会重新把任务队列中的所有任务执行完了再重新进入idle
 * 如果没有调度线程处理idle状态,那也就没有必要发通知了
 */
void IOManager::tickle()
{
    LOG(DEBUG, "tickle");
    if (!hasIdleThreads())
    {
        return;
    }
    int ret = write(m_tickleFds[1], "T", 1);
    SYLAR_ASSERT(ret == 1);
}

bool IOManager::stopping()
{
    uint64_t timeout = 0;
    return stopping(timeout);
}

// 判断是否可以停止,同时获取最近一个定时器的超时时间 timeout最近一个定时器的超时时间
// 用于idle协程的epoll_wait
bool IOManager::stopping(uint64_t &timeout)
{
    // 对于IOManager而言,必须等所有待调度等IO都执行完了才可以退出
    // 添加定时器功能后,还应该保证没有剩余的定时器待触发
    timeout = getNextTimer();
    return timeout == ~0ull && m_pendingEventCount == 0 && Scheduler::stopping();
}

/**
 * 调度器无调度会阻塞在idle协程上,对于IO调度而言,idle状态应该关注两件事,
 * 一是有没有新的调度任务,对应Schduler::schedule()
 * 如果有新的调度任务,那应该立即退出idle状态,并执行对应的任务,
 * 二是关注当前注册的所有IO事件有没有触发,如果有触发,那么应该执行
 * IO事件对应的回调函数
 */

/**
 * @brief idle协程
 * @details 对于IO协程调度来说,应阻塞在等待IO事件上,idle退出的时机是epoll_wait返回
 * 对应的操作是tickle或注册的IO事件就绪
 * 调度器无调度任务时会阻塞idle协程上,对IO调度器而言,idle状态应该关注两件事
 * 一是有没有新的调度任务,对应Schduler::schedule()
 * 如果有新的调度任务,那应该立即退出idle状态,并执行对应的任务
 * 二是关注当前注册的所有IO事件有没有触发,如果有触发,那么应该执行IO事件对应的回调函数
 * 
 * 定时器事件就加入到调度器里进行调度,如果是IO事件,就执行IO事件对应的回调函数
 */
void IOManager::idle()
{
    LOG(DEBUG, "idle");

    // 一次epoll_wait最多检测256个就绪事件,如果就绪事件超过了这个数,那么会在下轮epoll_wait继续处理
    const uint64_t MAX_EVENTS = 256;
    epoll_event *events = new epoll_event[MAX_EVENTS];
    std::shared_ptr<epoll_event> shared_events(events, [](epoll_event *ptr)
                                               { delete[] ptr; });

    while (true)
    {
        // 获取下一个定时器的超时时间,顺便判断调度器是否停止
        uint64_t next_timeout = 0;
        if (SYLAR_UNLIKELY(stopping(next_timeout)))
        {
            LOG(DEBUG, "name = %s idle stopping exit", getName());
            break;
        }

        // 阻塞在epoll_wait上,等待事件发生或定时器超时
        int ret = 0;
        do
        {
            // 默认超时时间5秒,如果下一个定时器的超时时间大于5秒,则仍以5秒来计算超时,避免定时器超时时间太大时,epoll_wait一直阻塞
            static const int MAX_TIMEOUT = 50000;
            if (next_timeout != ~0ull)
            {
                next_timeout = std::min((int)next_timeout, MAX_TIMEOUT);
            }
            else
            {
                next_timeout = MAX_TIMEOUT;
            }

            int ret = epoll_wait(m_epfd, events, MAX_EVENTS, (int)next_timeout);
            if (ret < 0 && errno == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        } while (true);

        // 收集所有已超时的定时器,执行回调函数
        std::vector<std::function<void()>> cbs;
        listExpireCb(cbs);
        if (!cbs.empty())
        {
            for (const auto &cb : cbs)
            {
                schedule(cb);
            }
            cbs.clear();
        }

        // 遍历所有发生的事件,根据epol_wait的私有指针找到对应的FdContext,进行事件处理
        for (int i = 0; i < ret; i++)
        {
            epoll_event &event = events[i];
            if (event.data.fd == m_tickleFds[0])
            {
                // ticklefd[0]用于通知协程调度,这时只需要把管道里的内容读完即可
                uint8_t dummy[256];
                while (read(m_tickleFds[0], dummy, sizeof(dummy)) > 0)
                    ;
                continue;
            }

            FdContext *fd_ctx = (FdContext *)event.data.ptr;
            FdContext::MutexType::Lock lock(fd_ctx->mutex);

            /**
             * EPOLLERR:出错,比如读写端已经关闭的pipe
             * EPOLLHUP:套接字对端关闭
             * 出现这两种情况,应该同时触发fd的读和写事件,否则有可能出现注册的事件永远得不到执行的情况
             */

            if (event.events & (EPOLLERR | EPOLLHUP))
            {
                event.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->events;
            }

            int real_events = NONE;
            if (event.events & EPOLLIN)
            {
                real_events |= READ;
            }
            if (event.events & EPOLLOUT)
            {
                real_events |= WRITE;
            }

            if ((fd_ctx->events & real_events) == NONE)
            {
                continue;
            }

            // 删除已经发生的事件,将剩下的事件重新加入epoll_wait
            int left_events = (fd_ctx->events & ~real_events);
            int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            event.events = EPOLLET | left_events;

            int ret2 = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
            if (ret2)
            {
                LOG(ERROR, "epoll_ctl( %d, %d, %d, %d,):%d, (%d)(%s)", m_epfd, (EpollCtlOp)op, fd_ctx->fd, (EPOLL_EVENTS)event.events, ret2, errno, strerror(errno));
            }

            // 处理已经发生的事件,也就是让调度器调度指定的函数或协程
            if (real_events & READ)
            {
                fd_ctx->triggerEvent(READ);
                --m_pendingEventCount;
            }

            if (real_events & WRITE)
            {
                fd_ctx->triggerEvent(WRITE);
                --m_pendingEventCount;
            }
        }

        /**
         * 一旦处理完所有的事件,idle协程yield,这样可以让协程调度(Scheduler::run)重新检查是否有新的调度任务
         * 上面的triggerEvent实际也只是把对应的fiber重新加入调度,要执行的话还要idle协程退出
         */

        Fiber::ptr cur = Fiber::GetThis();
        auto raw_ptr = cur.get();
        cur.reset();

        raw_ptr->yield();
    }
}

// 当有定时器插入到头部的时候,要重新更新epoll_wait的超时时间,这里唤醒idle协程以便于使用新的超时时间
void IOManager::onTimerInsertAtFront()
{
    tickle();
}